package com.openapi.protocol.decode.subPackageHandle;

import com.openapi.data.JT808PackageData;
import com.openapi.data.JT808PushData;
import com.openapi.protocol.encode.Package808;
import com.openapi.service.KafkaService;
import exsun.jt808.data.util.executorsThread.ScheduledHandle;
import exsun.jt808.data.distribute_data.jt808.GeneralReply;
import exsun.jt808.data.upload_data.Jt808EntityBase;
import exsun.jt808.data.util.parse.BitOperator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import com.openapi.config.JT808ServerConfig;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * 接收消息的分包处理
 * @param <T>
 */
public abstract class SubPackageHandle<T extends JT808PackageData>
{
    private static final Logger logger = LoggerFactory.getLogger(SubPackageHandle.class);
    private ConcurrentHashMap<String, ConcurrentHashMap<Integer, T>> _dics = new ConcurrentHashMap<>();

    protected static KafkaService kafkaService;

    @Autowired
    private void setKafkaService(KafkaService kafkaService) {
        SubPackageHandle.kafkaService = kafkaService;
    }

    protected static JT808ServerConfig jt808ServerConfig;

    @Autowired
    private void setJT808ServerConfig(JT808ServerConfig jt808ServerConfig) {
        SubPackageHandle.jt808ServerConfig = jt808ServerConfig;
    }

    public SubPackageHandle()
    {
        autoRemoveExprie();
    }

    private void autoRemoveExprie(){
        Runnable runnable = new Runnable() {
            public void run() {
                try {
                    for (String dictKey : _dics.keySet())
                    {
                        Map<Integer, T> datas = _dics.get(dictKey);

                        for (Integer dataKey : datas.keySet()){
                            T data = datas.get(dataKey);
                            if (new Date().getTime() - data.pte.getTime() > 1000*600)
                            {
                                remove(dictKey);
                                logger.info("[SubPackageHandle]clare expire，phoneNum:{0},time:{1}", data.dvo, data.pte.toString());
                            }
                            break;
                        }
                    }
                }catch (Exception e){
                    logger.error("SubpackageHandle AutoRemoveExprie:" + e.getMessage());
                }
            }
        };
        // 第二个参数为首次执行的延时时间，第三个参数为定时执行的间隔时间
        ScheduledHandle.getInst().getService().scheduleAtFixedRate(runnable, 0, 60, TimeUnit.SECONDS);
    }

    public abstract String getPakKey(T data);

    public abstract Jt808EntityBase parse(T data, byte[] totalBodys);

    public JT808PushData Response(T data){
        return Package808.generalReply_0x8001(new GeneralReply(data,0));
    }

    public SubPackageStatus append(T data)
    {
        addToCache(data);
        //无缺失项
        if (all(data)){
            return SubPackageStatus.complete;
        }
        if (data.packageIndex == data.getPackageSize()){
            return SubPackageStatus.missing;
        }else{
            return SubPackageStatus.receiving;
        }
    }

    private void addToCache(T data){
        String dickey = getPakKey(data);
        ConcurrentHashMap<Integer, T> dicTemp;
        if(!_dics.containsKey(dickey)){
            dicTemp = new ConcurrentHashMap<>();
        }else{
            dicTemp = _dics.get(dickey);
        }
        dicTemp.put(data.packageIndex, data);
        _dics.put(dickey, dicTemp);
    }

    public void remove(String dictKey){
        _dics.remove(dictKey);
    }

    /**
     * 获取当前缓存的数据包
     * @param data
     * @return
     */
    private ConcurrentHashMap<Integer, T> getLists(T data)
    {
        String dictKey = getPakKey(data);
        if (_dics.containsKey(dictKey)){
            return _dics.get(dictKey);
        }
        return new ConcurrentHashMap<>();
    }

    /**
     * 获取第一包
     * @param data
     * @return
     */
    public T getPakage(T data, int index)
    {
        String dictKey = getPakKey(data);
        if (!_dics.containsKey(dictKey))return null;
        if(!_dics.get(dictKey).containsKey(index))return null;
        return _dics.get(dictKey).get(index);
    }

    /**
     * 组包
     * @param data
     * @return
     */
    public byte[] getAllData(T data)
    {
//        if(!all(data))return null;
        ConcurrentHashMap<Integer, T> map = getLists(data);
        byte[] totalBytes = new byte[0];
        for(int n = 1; n <= data.packageSize; n++){
            if(!map.containsKey(n))return null;
            totalBytes = BitOperator.concatAll(totalBytes, map.get(n).getBodyBuffers());
        }
        return totalBytes;
    }

    /**
     * 获取缺失的包
     * @param data
     * @return
     */
    public List<Integer> getMissItems(T data)
    {
        List<Integer> list = new ArrayList<>();
        Map<Integer, T> multiItems = getLists(data);
        for (int i = 1; i < data.getPackageSize(); i++)
        {
            if (!multiItems.containsKey(i))
            {
                list.add(i);
            }
        }
        return list;
    }

    /**
     * 返回当前包是否接收完毕
     * @param data
     * @return
     */
    private boolean all(T data) {
        Map<Integer, T> map = getLists(data);
        Integer totalCount = data.getPackageSize();
        if (totalCount == null||totalCount == 0||
                totalCount == map.size()&& totalCount == Collections.max(map.keySet())) return true;
        return false;
    }

    public void setResultObj(Jt808EntityBase obj, T orgData, String topic){
        obj.dvo = orgData.dvo;
        obj.mid = orgData.getMid();
        obj.tpc = topic;
    }
}
